/*
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.ucr.cs.dblab.acestore.operators;

import java.nio.ByteBuffer;
import java.util.List;
import java.util.Vector;

import edu.ucr.cs.dblab.acestore.api.AbstractOperator;
import edu.ucr.cs.dblab.acestore.api.IOperator;
import edu.ucr.cs.dblab.acestore.api.ISinglePrevOperatorFactory;
import edu.ucr.cs.dblab.acestore.base.ACEContext;
import edu.ucr.cs.dblab.acestore.base.LRUCacheAlgorithm;
import edu.ucr.cs.dblab.acestore.base.datatypes.ISerDeserializer;
import edu.ucr.cs.dblab.acestore.base.datatypes.IntegerSerDeserializer;
import edu.ucr.cs.dblab.acestore.base.datatypes.SchemaDescriptor;

/**
 * A sort operator sorts the input data from its
 * previous operator, based on the given schema
 * and the sorting field. <br/>
 * 
 * The output of the operator
 * will be a series of pages containing sorted
 * data, with each page linking to the page after
 * it. <br/>
 * 
 * The final page is marked by -1 on the link
 * position. <br/>
 * 
 * @author jarodwen
 * @created Apr 28, 2011
 */
public class SortOperatorFactory implements ISinglePrevOperatorFactory {

    private final int pageLimits;
    private final SchemaDescriptor schemaDescriptor;
    private final int compareField;

    public SortOperatorFactory(int pageLimits,
            SchemaDescriptor schemaDescriptor, int compareField) {
        this.pageLimits = pageLimits;
        this.schemaDescriptor = schemaDescriptor;
        this.compareField = compareField;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * edu.ucr.cs.dblab.acestore.api.IOperatorFactory#createOperator(edu.ucr
     * .cs.dblab.acestore.base.ACEContext)
     */
    @Override
    public IOperator createOperator(final ACEContext ctx,
            final IOperator prevOperator) {
        return new AbstractOperator() {

            private final int idxID = ctx.allocateIndex();
            private List<Integer> runIDs;
            private ByteBuffer[] bufPages;
            private int workingBuf;
            private int sortedIdx;
            private int nextSortedPage = -1;
            private int sortedPageCount = 0;

            @Override
            public void open() throws Exception {
                this.runIDs = new Vector<Integer>();
                // Allocate pages for sorting
                this.bufPages = new ByteBuffer[pageLimits];
                for (int i = 0; i < pageLimits; i++) {
                    bufPages[i] = ByteBuffer.wrap(new byte[ctx.pageSize]);
                }
                this.workingBuf = 0;
                this.sortedIdx = -1;
                prevOperator.open();
            }

            @Override
            public void getNext(ByteBuffer page) throws Exception {
                if (this.sortedIdx < 0) {
                    partialSort();
                    merge();
                }
                if (nextSortedPage < 0) {
                    page.putInt(page.capacity() - INTSIZE - RESERVED_END, -1);
                } else {
                    System.arraycopy(ctx.readPage(sortedIdx, nextSortedPage)
                            .array(), 0, page.array(), 0, page.capacity());
                    nextSortedPage = page.getInt(page.capacity() - INTSIZE - RESERVED_END);
                }
                page.putInt(page.capacity() - RESERVED_END, sortedPageCount);
            }

            private void partialSort() throws Exception {

                ByteBuffer page = ByteBuffer.wrap(new byte[ctx.pageSize]);

                // Fetch pages from the previous operator
                prevOperator.getNext(page);

                while (true) {
                    if (workingBuf >= bufPages.length) {
                        // Sort buffers
                        // Get total number of records
                        int recCount = 0;
                        for (int i = 0; i < bufPages.length; i++) {
                            recCount += bufPages[i].getInt(bufPages[i]
                                    .capacity() - 2 * INTSIZE - RESERVED_END);
                        }
                        // Initialize links for sort
                        int[] links = new int[recCount * 2];
                        int linkIndex = 0;
                        for (int i = 0; i < bufPages.length; i++) {
                            int recIndex = bufPages[i].getInt(bufPages[i]
                                    .capacity() - 2 * INTSIZE - RESERVED_END);
                            for (int j = 0; j < recIndex; j++) {
                                links[linkIndex] = i;
                                links[linkIndex + 1] = j;
                                linkIndex += 2;
                            }
                        }
                        // Quick sort using links
                        links = quicksort(links, 0, links.length - 2);

                        // Write sorted records onto buffer
                        runIDs.add(writeRunFile(links));
                        // Reset buffers
                        workingBuf = 0;
                    }
                    if(page.getInt(page.capacity() - INTSIZE - RESERVED_END) < 0 && page.getInt(page.capacity() - INTSIZE * 2 - RESERVED_END) <= 0)
                        break;
                    System.arraycopy(page.array(), 0,
                            bufPages[workingBuf].array(), 0, page.capacity());
                    workingBuf++;
                    // Reset page
                    if (page.getInt(page.capacity() - INTSIZE - RESERVED_END) >= 0) {
                        page.position(0);
                        prevOperator.getNext(page);
                    } else {
                        break;
                    }
                }
                // Handle records left in the buffers
                if (workingBuf > 0) {
                    // Sort buffers
                    // Get total number of records
                    int recCount = 0;
                    for (int i = 0; i < workingBuf; i++) {
                        recCount += bufPages[i].getInt(bufPages[i].capacity()
                                - 2 * INTSIZE - RESERVED_END);
                    }
                    // Initialize links for sort
                    int[] links = new int[recCount * 2];
                    int linkIndex = 0;
                    for (int i = 0; i < workingBuf; i++) {
                        int recIndex = bufPages[i].getInt(bufPages[i]
                                .capacity() - 2 * INTSIZE - RESERVED_END);
                        for (int j = 0; j < recIndex; j++) {
                            links[linkIndex] = i;
                            links[linkIndex + 1] = j;
                            linkIndex += 2;
                        }
                    }
                    // Quick sort using links
                    links = quicksort(links, 0, links.length - 2);

                    // Write sorted records onto buffer
                    runIDs.add(writeRunFile(links));
                    // Reset buffers
                    workingBuf = 0;
                }
            }

            @Override
            public void close() throws Exception {
                // Empty pages
                this.bufPages = null;
                // Deallocate index
                for (int id : runIDs) {
                    ctx.deallocateIndex(id);
                }
                ctx.deallocateIndex(idxID);
            }

            /**
             * Write the sorted records into a run file, and return the index
             * number of the run file.
             * 
             * @param links
             * @return
             * @throws Exception
             */
            private int writeRunFile(int[] links) throws Exception {
                int runIdx = ctx.allocateIndex();
                int pageCount = 0;
                ByteBuffer page = ByteBuffer.wrap(new byte[ctx.pageSize]);
                int recInPage = 0;
                int prevEndOffsetInPage = -1;
                for (int i = 0; i < links.length; i = i + 2) {
                    ByteBuffer recBuffer = bufPages[links[i]];
                    int prevEndOffset = links[i + 1] == 0 ? -1 : recBuffer
                            .getInt(recBuffer.capacity() - 2 * INTSIZE
                                    - links[i + 1] * INTSIZE - RESERVED_END);
                    int recLength = recBuffer.getInt(recBuffer.capacity() - 2
                            * INTSIZE - (links[i + 1] + 1) * INTSIZE - RESERVED_END)
                            - prevEndOffset;
                    // Check whether the page is overflowing
                    page.position(prevEndOffsetInPage + 1);
                    if (page.position() + recLength >= page.capacity() - 2
                            * INTSIZE - recInPage * INTSIZE - RESERVED_END) {
                        page.putInt(page.capacity() - INTSIZE - RESERVED_END, pageCount + 1);
                        ctx.writePage(runIdx, pageCount, page);
                        pageCount++;
                        // Reset the record count in the page
                        recInPage = 0;
                        page.putInt(page.capacity() - 2 * INTSIZE - RESERVED_END, recInPage);
                        // Reset the page position
                        prevEndOffsetInPage = -1;
                        page.position(prevEndOffsetInPage + 1);
                        // Insert again
                        if (page.position() + recLength >= page.capacity() - 2
                                * INTSIZE - recInPage * INTSIZE - RESERVED_END) {
                            throw new Exception(
                                    "Failed to insert a record into the buffer during sorting.");
                        }
                    }
                    page.put(recBuffer.array(), prevEndOffset + 1, recLength);
                    recInPage++;
                    page.putInt(page.capacity() - 2 * INTSIZE - RESERVED_END, recInPage);
                    page.putInt(page.capacity() - 2 * INTSIZE - recInPage
                            * INTSIZE - RESERVED_END, prevEndOffsetInPage + recLength);
                    prevEndOffsetInPage = prevEndOffsetInPage + recLength;
                }
                if (recInPage > 0) {
                    page.putInt(page.capacity() - INTSIZE - RESERVED_END, -1);
                    ctx.writePage(runIdx, pageCount, page);
                }
                // return the index
                return runIdx;
            }

            private void merge() throws Exception {
                if(runIDs.size() <= 0)
                    return;
                // Initialize the output page
                ByteBuffer outPage = ByteBuffer.wrap(new byte[ctx.pageSize]);
                // Do merge
                while (runIDs.size() >= pageLimits) {
                    int partialRunID = ctx.allocateIndex();
                    int partialRunPageID = 0;
                    // merge run files
                    int[] workingRecIdx = new int[bufPages.length];
                    for (int i = 0; i < bufPages.length; i++) {
                        workingRecIdx[i] = 0;
                        bufPages[i] = ctx.readPage(runIDs.get(i),
                                workingRecIdx[i]);
                    }
                    while (true) {
                        int minBuf = -1;
                        for (int i = 0; i < bufPages.length; i++) {
                            if (workingRecIdx[i] < 0)
                                // Skip exhausted page
                                continue;
                            if (workingRecIdx[i] >= bufPages[i]
                                    .getInt(bufPages[i].capacity() - 2
                                            * INTSIZE - RESERVED_END)) {
                                // Try to load new page
                                int nextPageID = bufPages[i].getInt(bufPages[i]
                                        .capacity() - INTSIZE - RESERVED_END);
                                if (nextPageID >= 0) {
                                    bufPages[i] = ctx.readPage(runIDs.get(i),
                                            nextPageID);
                                    workingRecIdx[i] = 0;
                                } else {
                                    workingRecIdx[i] = -1;
                                    continue;
                                }
                            }
                            if (minBuf < 0)
                                // Initialize minBuf value
                                minBuf = i;
                            else if (compareRecords(bufPages[minBuf],
                                    workingRecIdx[minBuf], bufPages[i],
                                    workingRecIdx[i]) > 0) {
                                // Do comparison
                                minBuf = i;
                            }
                        }
                        if (minBuf < 0)
                            break;
                        else {
                            // Write the record output into a new run file
                            int startOffset = workingRecIdx[minBuf] == 0 ? 0
                                    : bufPages[minBuf].getInt(bufPages[minBuf]
                                            .capacity()
                                            - 2
                                            * INTSIZE
                                            - workingRecIdx[minBuf] * INTSIZE - RESERVED_END) + 1;
                            int length = bufPages[minBuf]
                                    .getInt(bufPages[minBuf].capacity() - 2
                                            * INTSIZE
                                            - (workingRecIdx[minBuf] + 1)
                                            * INTSIZE - RESERVED_END)
                                    - startOffset + 1;
                            int recCountOut = outPage.getInt(outPage.capacity()
                                    - 2 * INTSIZE - RESERVED_END);
                            if (outPage.position() + length >= outPage
                                    .capacity()
                                    - 2
                                    * INTSIZE
                                    - (recCountOut + 1) * INTSIZE - RESERVED_END) {
                                // Update the page reference to next new page
                                outPage.putInt(outPage.capacity() - INTSIZE - RESERVED_END,
                                        partialRunPageID + 1);
                                // Write outPage into the run file if
                                // overflowing
                                ctx.writePage(partialRunID, partialRunPageID,
                                        outPage);
                                // Update page count
                                partialRunPageID++;
                                // Reset the output page
                                outPage.putInt(
                                        outPage.capacity() - 2 * INTSIZE - RESERVED_END, 0);
                                recCountOut = 0;
                                outPage.position(0);
                            }
                            // Insert into the outPage
                            outPage.put(bufPages[minBuf].array(), startOffset,
                                    length);
                            // Insert end offset
                            outPage.putInt(outPage.capacity() - 2 * INTSIZE
                                    - (recCountOut + 1) * INTSIZE - RESERVED_END,
                                    outPage.position() - 1);
                            // Update record count
                            outPage.putInt(outPage.capacity() - 2 * INTSIZE - RESERVED_END,
                                    recCountOut + 1);
                            // Update record index for this buffer
                            workingRecIdx[minBuf]++;
                        }
                    }
                    // Handle records left
                    if (outPage.getInt(outPage.capacity() - 2 * INTSIZE - RESERVED_END) > 0) {
                        outPage.putInt(outPage.capacity() - INTSIZE - RESERVED_END, -1);
                        ctx.writePage(partialRunID, partialRunPageID, outPage);
                        partialRunPageID++;
                        outPage.putInt(outPage.capacity() - 2 * INTSIZE - RESERVED_END, 0);
                        outPage.position(0);
                    }
                    for(int i = 0; i < bufPages.length; i++){
                        ctx.deallocateIndex(runIDs.remove(0));
                    }
                    runIDs.add(partialRunID);
                }
                // Final merge
                // Allocate index for sorted result
                this.sortedIdx = ctx.allocateIndex();
                int sortedWorkingPage = 0;
                int[] workingRecIdx = new int[runIDs.size()];
                for (int i = 0; i < runIDs.size(); i++) {
                    workingRecIdx[i] = 0;
                    bufPages[i] = ctx.readPage(runIDs.get(i), workingRecIdx[i]);
                }
                // Do final merge
                while (true) {
                    int minBuf = -1;
                    for (int i = 0; i < runIDs.size(); i++) {
                        if (workingRecIdx[i] < 0)
                            // Skip exhausted page
                            continue;
                        if (workingRecIdx[i] >= bufPages[i].getInt(bufPages[i]
                                .capacity() - 2 * INTSIZE - RESERVED_END)) {
                            // Try to load new page
                            int nextPageID = bufPages[i].getInt(bufPages[i]
                                    .capacity() - INTSIZE - RESERVED_END);
                            if (nextPageID >= 0) {
                                // A new page is loaded
                                bufPages[i] = ctx.readPage(runIDs.get(i),
                                        nextPageID);
                                workingRecIdx[i] = 0;
                            } else {
                                // All pages have been exhausted
                                workingRecIdx[i] = -1;
                                continue;
                            }
                        }
                        if (minBuf < 0)
                            // Initialize minBuf value
                            minBuf = i;
                        else if (compareRecords(bufPages[minBuf],
                                workingRecIdx[minBuf], bufPages[i],
                                workingRecIdx[i]) >= 0) {
                            // Do comparison
                            minBuf = i;
                        }
                    }
                    if (minBuf < 0)
                        break;
                    else {
                        // Write the record output into a new run file
                        int startOffset = workingRecIdx[minBuf] == 0 ? 0
                                : bufPages[minBuf].getInt(bufPages[minBuf]
                                        .capacity()
                                        - 2
                                        * INTSIZE
                                        - workingRecIdx[minBuf] * INTSIZE - RESERVED_END) + 1;
                        int length = bufPages[minBuf].getInt(bufPages[minBuf]
                                .capacity()
                                - 2
                                * INTSIZE
                                - (workingRecIdx[minBuf] + 1) * INTSIZE - RESERVED_END)
                                - startOffset + 1;
                        int recCountOut = outPage.getInt(outPage.capacity() - 2
                                * INTSIZE - RESERVED_END);
                        if (outPage.position() + length >= outPage.capacity()
                                - 2 * INTSIZE - (recCountOut + 1) * INTSIZE - RESERVED_END) {
                            // Write outPage into the run file if overflowing
                            outPage.putInt(outPage.capacity() - INTSIZE - RESERVED_END,
                                    sortedWorkingPage + 1);
                            ctx.writePage(this.sortedIdx, sortedWorkingPage,
                                    outPage);
                            sortedWorkingPage++;
                            outPage.putInt(outPage.capacity() - 2 * INTSIZE - RESERVED_END, 0);
                            recCountOut = 0;
                            outPage.position(0);
                        }
                        // Insert into the outPage
                        outPage.put(bufPages[minBuf].array(), startOffset,
                                length);
                        // Insert record end offset
                        outPage.putInt(outPage.capacity() - 2 * INTSIZE
                                - (recCountOut + 1) * INTSIZE - RESERVED_END,
                                outPage.position() - 1);
                        // Update record count
                        outPage.putInt(outPage.capacity() - 2 * INTSIZE - RESERVED_END,
                                recCountOut + 1);
                        // Update record index for this buffer
                        workingRecIdx[minBuf]++;
                    }
                }
                // Handle records left
                if (outPage.getInt(outPage.capacity() - 2 * INTSIZE - RESERVED_END) > 0) {
                    outPage.putInt(outPage.capacity() - INTSIZE - RESERVED_END, -1);
                    ctx.writePage(this.sortedIdx, sortedWorkingPage, outPage);
                }
                this.sortedPageCount = sortedWorkingPage + 1;
                nextSortedPage = 0;
            }

            /***********************************
             * Quick sort functions
             ***********************************/

            /**
             * Quick function.
             * 
             * Partition the whole data set into two parts using the pivot
             * value, and do quick sort on the two parts separately.
             * 
             * @param links
             * @param low
             * @param high
             * @return
             * @throws Exception
             */
            private int[] quicksort(int[] links, int low, int high)
                    throws Exception {
                if (high > low) {
                    int partitionPivotIndex = (int) (Math.random() * (high - low))
                            / 2 * 2 + low;
                    int newPivotIndex = partition(links, low, high,
                            partitionPivotIndex);
                    quicksort(links, low, newPivotIndex - 2);
                    quicksort(links, newPivotIndex + 2, high);
                }
                return links;
            }

            private int partition(int[] array, int lo, int hi, int pivotIndex)
                    throws Exception {

                swap(array, pivotIndex, hi); // send to the back

                int index = lo;

                for (int i = lo; i < hi; i = i + 2) {
                    if (compareLinks(array, i, hi) <= 0) {
                        swap(array, i, index);
                        index = index + 2;
                    }
                }

                swap(array, hi, index);

                return index;
            }

            private void swap(int[] array, int i, int j) {
                int temp1 = array[i];
                int temp2 = array[i + 1];
                array[i] = array[j];
                array[i + 1] = array[j + 1];
                array[j] = temp1;
                array[j + 1] = temp2;
            }

            private int compareLinks(int[] array, int i, int j)
                    throws Exception {
                return compareRecords(bufPages[array[i]], array[i + 1],
                        bufPages[array[j]], array[j + 1]);
            }

            private int compareRecords(ByteBuffer b1, int idx1, ByteBuffer b2,
                    int idx2) throws Exception {
                int offset1 = idx1 == 0 ? 0 : b1.getInt(b1.capacity() - 2
                        * INTSIZE - idx1 * INTSIZE - RESERVED_END) + 1;
                int offset2 = idx2 == 0 ? 0 : b2.getInt(b2.capacity() - 2
                        * INTSIZE - idx2 * INTSIZE - RESERVED_END) + 1;
                int fieldCount1 = b1.getInt(offset1);
                int fieldCount2 = b2.getInt(offset2);
                int len = schemaDescriptor.getFields()[compareField]
                        .getTypeSize();

                return schemaDescriptor.getFields()[compareField].compare(
                        b1.array(), offset1 + INTSIZE + fieldCount1 * INTSIZE
                                + compareField * INTSIZE, len, b2.array(),
                        offset2 + INTSIZE + fieldCount2 * INTSIZE
                                + compareField * INTSIZE, len);

            }

        };
    }

    public static void main(String[] args) throws Exception {
        ACEContext ctx = new ACEContext(100, 3, new LRUCacheAlgorithm());
        FileScanOperatorFactory scanOperatorFactory = new FileScanOperatorFactory(
                "/Users/jarodwen/Desktop/hyracks/data/simple2D.txt",
                new SchemaDescriptor(new ISerDeserializer[] {
                        IntegerSerDeserializer.getSingleton(),
                        IntegerSerDeserializer.getSingleton() }), ",");
        IOperator scanOperator = scanOperatorFactory.createOperator(null, null);
        SortOperatorFactory sortOperatorFactory = new SortOperatorFactory(3,
                new SchemaDescriptor(new ISerDeserializer[] {
                        IntegerSerDeserializer.getSingleton(),
                        IntegerSerDeserializer.getSingleton() }), 0);
        IOperator sortOperator = sortOperatorFactory.createOperator(ctx,
                scanOperator);
        TextPlainOutputOperatorFactory printOperatorFactory = new TextPlainOutputOperatorFactory(
                new SchemaDescriptor(new ISerDeserializer[] {
                        IntegerSerDeserializer.getSingleton(),
                        IntegerSerDeserializer.getSingleton() }), "\t");
        IOperator printOperator = printOperatorFactory.createOperator(ctx,
                sortOperator);
        ByteBuffer page = ByteBuffer.wrap(new byte[100]);
        page.position(0);
        printOperator.open();
        printOperator.getNext(page);
        while (page.getInt(page.capacity() - 4 - RESERVED_END) >= 0) {
            page.position(0);
            printOperator.getNext(page);
        }
        printOperator.close();
    }

}
